package edu.gmu.middleware;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.ArrayList;
import java.util.List;

/**
 * UDPStreamSink to System Resource Manager provider.  This Sink
 * uses Stream based IO.  It supports both only UDP.
 *
 * @author  James H. Pope
 * @version $Revision:$ $Date:$
 */
public class UDPChannelSink implements Sink
{
    /**
     * Default buffer sizes for UDP Datagrams is 8192 - theoretical
     * is 65,507 (IPv4) or 65536 (IPv6)
     */
    protected static final int DEFAULT_BUFFER_SIZE = 65536;
    
    /**
     * Maximum size a Datagram can be (IP header is at least 20 bytes
     * plus 8 byte UDP header - 2 byte length includes them as well).
     */
    protected static final int MAX_IPV4_PACKET_SIZE = 65507;
    
    //----------------------------- Sink Variables ---------------------------//
    private DatagramChannel datagramChannel = null;
    
    private SocketAddress remoteAddress = null;
    private SocketAddress localAddress = null;
    private int bufferSize = DEFAULT_BUFFER_SIZE;
    
    protected volatile boolean opened = false;
    protected volatile boolean closed = false;
    
    /**
     * Creates a new UDPChannelSink with specified values.
     * @param remoteAddress
     * @param localAddress
     */
    public UDPChannelSink(  SocketAddress remoteAddress,
                            SocketAddress localAddress )
    {
        this( remoteAddress, localAddress, DEFAULT_BUFFER_SIZE );
    }
    
    /**
     * Creates a new UDPChannelSink with specified values.
     * @param DatagramSocket datagramSocket
     * @param remoteAddress
     * @param localAddress
     * @param bufferSize
     */
    public UDPChannelSink(  SocketAddress remoteAddress,
                            SocketAddress localAddress,
                            int bufferSize )
    {
        this.remoteAddress = remoteAddress;
        this.localAddress  = localAddress;
        this.bufferSize    = bufferSize;
    }
    
    /**
     * Determones if Sink is open for reading/writing.
     * @return true if connected, false otherwise
     */
    public boolean isConnected()
    {
        return (this.closed == false && this.opened == true);
    }
    
    /**
     * Opens the Sink for reading from or writing to.
     */
    public void connect() throws IOException
    {
        this.datagramChannel = DatagramChannel.open();
        
        //this.datagramChannel.connect(this.remoteAddress);
        this.datagramChannel.socket().setSendBufferSize(bufferSize);
        this.datagramChannel.socket().setReceiveBufferSize(bufferSize);
        
        // Bind to local port - clients really dont need this
        this.datagramChannel.socket().bind(localAddress);
        
        
        // "Connect" to remoteAddress - eliminates subsequent security checks
        /*
         * We are not "connected" so that if the server goes down, we
         * continue to operate - when the server comes back up, we still
         * have the socket in good order (i.e. no "ICMP Not
         * Reachable" exception as would occur if "connected")
         */
        
        this.opened = true;
    }
    
    
    /**
     * Stops the Dispatcher from sending/receiving. Also results in
     * the sink being closed.
     */
    public void disconnect()
    {
        // Set boolean state for reading thread
        this.closed = true;
        
        if( this.datagramChannel != null )
        {
            try
            {
                this.datagramChannel.close();
                //this.datagramSocket = null;
            }
            catch( IOException ioe )
            {
                //ignore
            }
        }
    }
    
    
    
    //----------------------------- Sink Methods -----------------------------//
    
    /**
     * Reads the packet from the Sink (usually a network Socket).
     * This method blocks until data is ready.
     * @param in
     * @return number of bytes read
     */
    public Header read( ByteBuffer in ) throws IOException
    {   
        // Check State and complain if not good
        if( this.closed == true || this.opened == false )
        {
            throw new IllegalStateException( "Sink not in state to send " + 
                    "- closed=" + closed + " opened=" + opened );
        }
        
        /*
         * Read bytes from socket - As this is blocking io, this reads
         * all the bytes sent in the datagram - they all have to be present
         */
        in.clear();
        
        SocketAddress remoteClient = this.datagramChannel.receive( in );

        int lengthReceived = in.position();
        //System.out.println( "lengthReceived " + this.incoming.position() );
        
        // Now copy from incoming buffer to new buffer to return
        in.flip();
        
        /*
         * Peek at the header
         */
        Header header = new Header();
        header.unmarshal(in);
        in.rewind();
        
        //System.out.println( "Received="+lengthReceived );
        
        return header;
    }
    
    /**
     * Writes the packet to the Sink.  Though not as
     * significant as a read, this call is also blocking.
     * @param out
     * @return number of bytes sent
     */
    public int write( ByteBuffer out ) throws IOException
    {   
        // Check State and complain if not good
        if( this.closed == true || this.opened == false )
        {
            throw new IllegalStateException( "Sink not in state to send " + 
                    "- closed=" + closed + " opened=" + opened );
        }
        
        // Socket sends from 0 to data length to our host/port
        
        //System.out.println( "Writing " + out.position() );
        
        out.flip();
        //System.out.println( "Writing="+Packet.toString(outgoing) );
        int written = this.datagramChannel.send(out, remoteAddress);
        out.clear();
        
        // If all went well - assume bytes written was length of message
        return written;
    }
    
    
    
    
    
    
    
    //---------------------------- Event Methods -----------------------------//
    //
    //------------------------------------------------------------------------//
    
    private volatile boolean running    = false;
    private List sinkListeners          = new ArrayList();
    private List readerThreads          = null;
    
    
    private Thread makeThread( String name, final ByteBuffer in )
    {
        Runnable runnable = new Runnable()
        {
            public void run()
            {
                System.out.println("Buffer capacity " + in.capacity());
                running = true;
                while( running == true )
                {
                    
                    try
                    {
                        // Critical design decision concerning in buffer
                        // Each reader thread has it's own
                        Header header = read( in );
                        //System.out.println("SinkManager Received " + packet);
                        if( running == true )
                        {
                            // Callbacks (e.g. Dispatcher) may take some time
                            
                            /*
                             * Notifies listeners interested in serialized
                             * messages - e.g. recorders and routers.
                             */
                            fireDataReceived(header, in);
                        }
                    }
                    catch( Exception e )
                    {
                        //System.out.println("WE GOT ABNORMAL !!!!!!!!!");
                        // If our stop has been called - exit thread
                        if( running == true )
                        {
                            /*
                             * Catch io and user callback exceptions
                             * Note: Fairly dubious to allow clients
                             * to properly handle exceptions???
                             */
                            fireDisconnected(in,e);
                        }
                        // else ignore - we are stopping
                    }
                }
                // Should we stop running - close the sink
                running = false; // by definition, we are close enough to done
                disconnect();
                System.out.println("###############################");
                System.out.println("# Reader is exitting - bye!");
                System.out.println("###############################");
            }
        };
        // Overwrites any previous thread - are we reentrant?
        Thread t = new Thread(runnable);
        //thread.setDaemon(true);
        t.setName( name );
        
        return t;
    }
    
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param in
     * @param e
     */
    public void fireDisconnected( ByteBuffer in, Exception e )
    {
        //Do this in order to only create Event once
        SinkDisconnectedEvent event = null;
        
        // Determine type of event based on state
        int type = SinkDisconnectedEvent.NORMAL;
        if( this.isConnected() )
        {
            type = SinkDisconnectedEvent.ABNORMAL;
        }

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.sinkListeners.size(); i++)
        {
            // lazy initialization
            if(event == null)
            {
                event = new SinkDisconnectedEvent(  this,
                                            (long)System.currentTimeMillis(),
                                            type,
                                            in,
                                            this,
                                            e
                                     );
            }
            // Get listener
            SinkListener listener = ((SinkListener)this.sinkListeners.get(i));
            // Call listener
            listener.sinkDisconnected( event );
        }
    }
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param header
     * @param in
     */
    public void fireDataReceived( Header header, ByteBuffer in )
    {
        //Do this in order to only create Event once
        SinkDataEvent event = null;

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.sinkListeners.size(); i++)
        {
            // lazy initialization
            if(event == null)
            {
                event = new SinkDataEvent(  this,
                                        (long)System.currentTimeMillis(),
                                        header,
                                        in,
                                        this
                                     );
            }
            // Get listener
            SinkListener listener = ((SinkListener)this.sinkListeners.get(i));
            // Call listener
            listener.dataReceived( event );
        }
    }
    
    /**
     * Adds to list of listeners. First checks to see if
     * listener is already in list, if it is it does not add it a second time.
     * @param listener to add.
     */
    public void addSinkListener( SinkListener listener )
    {
        if( !this.sinkListeners.contains(listener) )
        {
            this.sinkListeners.add(listener);
        }
    }

    /**
     * Removes from list of listeners.
     * @param listener to remove.
     */
    public void removeSinkListener( SinkListener listener )
    {
        this.sinkListeners.remove(listener);
    }
    
    

}
